package com.ruoyi.config.mqtt;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.domain.DeviceData;
import com.ruoyi.service.DeviceDataShardingService;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Map;

@Configuration
public class MqttConfig {

    @Value("${mqtt.host:110.41.18.202}")
    private String mqttHost;

    @Value("${mqtt.port:1883}")
    private int mqttPort;

    @Value("${mqtt.username:mqtt_test}")
    private String mqttUsername;

    @Value("${mqtt.password:yuanzihuan@0512}")
    private String mqttPassword;

    @Autowired
    private DeviceDataShardingService deviceDataShardingService;

    private final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * MQTT客户端工厂
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{"tcp://" + mqttHost + ":" + mqttPort});
        options.setUserName(mqttUsername);
        options.setPassword(mqttPassword.toCharArray());
        options.setCleanSession(true);
        options.setKeepAliveInterval(60);
        options.setConnectionTimeout(30);
        factory.setConnectionOptions(options);
        
        System.out.println("MQTT连接配置 - 服务器: tcp://" + mqttHost + ":" + mqttPort);
        System.out.println("MQTT连接配置 - 用户名: " + mqttUsername);
        System.out.println("MQTT连接配置 - 密码: " + (mqttPassword != null ? "[已设置]" : "[未设置]"));
        
        return factory;
    }

    /**
     * MQTT输入通道
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT消息驱动适配器
     */
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter("ruoyi-consumer", mqttClientFactory(),
                        "testtopic/temphum");
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    /**
     * MQTT消息处理器
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                try {
                    String payload = message.getPayload().toString();
                    String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
                    
                    System.out.println("收到MQTT消息 - 主题: " + topic + ", 内容: " + payload);
                    
                    // 解析JSON数据
                    Map<String, Object> data = objectMapper.readValue(payload, Map.class);
                    
                    // 创建DeviceData对象
                    DeviceData deviceData = new DeviceData();
                    deviceData.setSensorId(data.get("sensorId").toString());
                    deviceData.setVersion(data.get("version").toString());
                    deviceData.setTemperature(Double.valueOf(data.get("temperature").toString()));
                    deviceData.setHumidity(Double.valueOf(data.get("humidity").toString()));
                    
                    // 处理时间戳 - 修复：使用毫秒级时间戳
                    // 处理时间戳 - 添加严格的时间戳验证
                    String collectTimeStr = data.get("collectTime").toString();
                    long timestamp = Long.parseLong(collectTimeStr);
                    long currentTime = System.currentTimeMillis();
                    
                    System.out.println("当前系统时间戳: " + currentTime);
                    System.out.println("接收到的时间戳: " + timestamp);
                    System.out.println("时间差: " + (timestamp - currentTime) + "ms");
                    
                    // 检查时间戳是否在合理范围内（允许前后1小时的误差）
                    long oneHourInMillis = 60 * 60 * 1000;
                    if (Math.abs(timestamp - currentTime) > oneHourInMillis) {
                        System.err.println("警告：时间戳异常，超出合理范围，使用当前时间替代");
                        System.err.println("异常时间戳: " + timestamp + " -> " + 
                            LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()));
                        timestamp = currentTime;
                    }
                    
                    LocalDateTime collectTime = LocalDateTime.ofInstant(
                            Instant.ofEpochMilli(timestamp), ZoneId.systemDefault());
                    deviceData.setCollectTime(collectTime);
                    
                    System.out.println("最终使用的时间: " + collectTime);
                    System.out.println("对应的分表名: device_data_" + 
                        collectTime.toLocalDate().format(DateTimeFormatter.ofPattern("yyyyMMdd")));
                    
                    // 保存到数据库
                    deviceDataShardingService.saveDeviceData(deviceData);
                    
                    System.out.println("MQTT数据已保存到数据库: " + deviceData);
                    
                } catch (Exception e) {
                    System.err.println("处理MQTT消息失败: " + e.getMessage());
                    e.printStackTrace();
                }
            }
        };
    }
}